// Copyright 2023 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "InterThreadComm.h"
#include "Include/msghead.h"
#include "Log/logging.h"

using namespace std;

/**
 * @brief: construct
 * @param [in] multiEnq: multiEnq
 * @param [in] multiDeq: multiDeq
 * @param [in] block: queue is blocked
 */
/*lint -save -e1401 */
InterThreadComm::InterThreadComm(bool multiEnq, bool multiDeq, bool block)
    : m_packetDeque(),
      m_lock(),
      m_conditionVariable()
{
    m_multiEnq = multiEnq;
    m_multiDeq = multiDeq;
    m_block = block;
    m_status = true;
    m_totalDataSize = 0;
}
/*lint -restore */

/**
 * @brief: deconstruct
 */
InterThreadComm::~InterThreadComm()
{
    m_status = false;
    Clear();
}

/**
 * @brief: clear queue and free memory
 */
void InterThreadComm::Clear()
{
    void* packet = nullptr;

    unique_lock<mutex> lck(m_lock);
    m_conditionVariable.notify_all();
    while (!(m_packetDeque.empty())) {
        packet = m_packetDeque.front();
        m_packetDeque.pop_front();
        VmiFreeBuffer(packet);
    }
}

/**
 * @brief: If you want to release the communication object between threads, you should
 *         call this interface first, especially in blocking mode, and release the object
 *         after the blocked receiving thread exits.
 */
void InterThreadComm::Cancel()
{
    m_status = false;
    m_conditionVariable.notify_all();
}

/**
 * @brief: put data into data queue
 * @param [in] packet: data
 */
void InterThreadComm::PutPkt(const VmiBuffer& packet)
{
    uint8_t *buffer = packet.GetPointer();
    size_t bufferSize = packet.GetSize();
    if (buffer == nullptr || bufferSize <= sizeof(StreamMsgHead)) {
        ERR("error: bad packet to put, buffer:%p, size:%zu", buffer, bufferSize);
        VmiFreeBuffer(buffer); // VmiFreeBuffer会对buffer判空
        return;
    }

    unique_lock<mutex> lck(m_lock);
    if (m_status) {
        StreamMsgHead *head = reinterpret_cast<StreamMsgHead *>(buffer);
        if (head->getPayloadSize() == 0) {
            ERR("error: payload size is 0");
            VmiFreeBuffer(buffer);
            return;
        }
        m_totalDataSize += head->getPayloadSize();
        m_packetDeque.push_back(buffer);
        if (m_block) {
            m_conditionVariable.notify_one();
        }
    } else {
        VmiFreeBuffer(buffer);
    }
}

/**
 * @brief: get a data from queue
 * @return : pointer of data packet, or nullptr if there are no data packet in the queue, caller need to check it
 */
uint8_t* InterThreadComm::GetNextPkt()
{
    uint8_t* packet = nullptr;
    unique_lock<mutex> lck(m_lock);
    if (m_status) {
        if (m_block) {
            m_conditionVariable.wait(
                lck, [this]() -> bool { return !m_status || !m_packetDeque.empty(); });
        }
        if (m_status && !m_packetDeque.empty()) {
            packet = m_packetDeque.front();
            m_packetDeque.pop_front();
            StreamMsgHead* head = reinterpret_cast<StreamMsgHead *>(packet);
            m_totalDataSize -= head->getPayloadSize();
        }
    }
    return packet;
}

/**
 * @brief: get a data from queue with timeout
 * @param [in] timeoutMillis: unit is ms
 * @return : pointer of data packet, or nullptr if there are no data packet in the queue, caller need to check it
 */
uint8_t* InterThreadComm::GetNextPktWait(int timeoutMillis)
{
    uint8_t* packet = nullptr;
    unique_lock<mutex> lck(m_lock);
    if (m_status) {
        if (m_block) {
            m_conditionVariable.wait_for(lck, std::chrono::milliseconds(timeoutMillis),
                [this]() -> bool { return !m_status || !m_packetDeque.empty(); });
        }
        if (m_status && !m_packetDeque.empty()) {
            packet = m_packetDeque.front();
            m_packetDeque.pop_front();
            StreamMsgHead* head = reinterpret_cast<StreamMsgHead *>(packet);
            m_totalDataSize -= head->getPayloadSize();
        }
    }

    return packet;
}

/**
 * @brief: get queue size
 * @return : number of items in the queue
 */
size_t InterThreadComm::GetNumItems()
{
    unique_lock<mutex> lck(m_lock);
    return m_packetDeque.size();
}

/**
 * @brief: get total size
 * @return : total data size
 */
uint32_t InterThreadComm::GetTotalSize()
{
    return m_totalDataSize;
}
